import com.software.util.HBaseTools
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object ReadTest {
  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args:Array[String]):Unit={
    //1.创建Spark环境配置对象
    val conf =new SparkConf().setAppName("SparkSqlCSVExample").setMaster("local").set("spark.testing.memory", "2147480000")
    //2.创建SparkSession对象
    val spark:SparkSession=SparkSession.builder().config(conf).getOrCreate()
    val readHbase = getGlobalTempByCountryDataFrame(spark)
    readHbase.show(10)



  }

  /**
   * 读取HBase中globalLandTemperaturesByCountry表数据，并注册为临时表
   *
   * @param session
   * @return
   */
  def getGlobalTempByCountryDataFrame(session: SparkSession): DataFrame = {

    val hconf = HBaseTools.getHbaseConn("example_table3")
    val rdd: RDD[(ImmutableBytesWritable, Result)] =
      session.sparkContext.newAPIHadoopRDD(hconf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])
    import session.implicits._

    val df: DataFrame = rdd.map(f => {
      val result: Result = f._2
      //      val dt = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("dt")))
      val type1 = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("type")))
      val date1 = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("date")))
      val hour1 = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("hour")))
      //      val TJ = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("天津")))
      //      val SJZ = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("石家庄")))
      //      val TS = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("唐山")))
      //      val QHD = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("秦皇岛")))
      //      val HD = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("邯郸")))
      //      val Country = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("country")))
      // 转换成RDD[Row]

      (type1, date1, hour1)

    }).toDF("type","date","hour")
    df.show(false)
    //3.注册临时表
    df.createOrReplaceTempView("tbl_air")
    df
  }


}